package edu.cornell.cs.lsi.mapreduce.pass1.test;

import static edu.cornell.cs.lsi.mapreduce.Constants.G;
import static edu.cornell.cs.lsi.mapreduce.Constants.Lmax_SQUARED;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.FileWriter;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import edu.cornell.cs.lsi.mapreduce.Constants;



public class BruteForceTester {
	
	public static void main(String[] args) throws Exception{
		
		String inputDataPrefix = "C:\\cygwin\\bin\\hadoop-0.20.2\\bin\\";
		String inputCalcPrefix = "C:\\cygwin\\bin\\hadoop-0.20.2\\bin\\outdir\\";
		//String inputCalcPrefix = "C:\\Users\\David\\Desktop\\";
		
		String inputDataFileName = "test3.txt";
		String inputCalcFileName = "part-r-00000";
		
		String outputPath = "C:\\cygwin\\out.txt";
		
		FileWriter writer = new FileWriter(outputPath);
		
		
		boolean doComp = false;
		
		String inputDataPath = inputDataPrefix + inputDataFileName;
		String inputCalcPath = inputCalcPrefix + inputCalcFileName;
		
				
		List<DPoint> allPoints = getPointsFromDataFiles(inputDataPath);
		
		
		int edgeCount = getCorEdgeCount(allPoints);
		
		Map<Integer, List<DPoint>> colMap = getColMap(allPoints);
		
		List<Link> calcLinks = null;
		
		if(doComp){
			calcLinks = getCalcLinks(inputCalcPath);
			
			printLinkList(calcLinks);
			
			System.out.println("\n\n------------\n");
		}
		
		List<Link> corLinks = getCorLinks(colMap);
		
		//printLinkList(corLinks);
		
		if(doComp) compareLinks(calcLinks, corLinks, edgeCount);
		
		System.out.println(edgeCount);
		
		for(Link l : corLinks){
			writer.append(l.toString() + "\n");
		}
		writer.flush();
		writer.close();
		
	}

	private static void printLinkList(List<Link> links){
		for(Link l : links){
			System.out.println(l);
		}
	}
	
	
	private static void compareLinks(List<Link> calcLinks, List<Link> corLinks, int corEdgeCount){
		
		int i;
		Link curCalcLink;
		Link curCorLink;
		int calcEdgeCount = 0;
		
		
		
		if(calcLinks.size() != corLinks.size()){
			System.out.println("Failure: wrong size");
			return;
		}
		
		for(i = 0; i < calcLinks.size(); i++){
			curCalcLink = calcLinks.get(i);
			curCorLink = corLinks.get(i);
			if(!curCalcLink.equals(curCorLink)){
				System.out.println("Failure: unequal");
				return;
			}
		}
		/*
		for(Link l : calcLinks){
			if(l.edgeCount > -1){
				calcEdgeCount += l.edgeCount;
			}
		}
		
		if(calcEdgeCount != corEdgeCount){
			System.out.println(String.format("Edge count mismatch. Expected %d, got %d.", corEdgeCount, calcEdgeCount));
			return;
		}*/
		
		System.out.println("Pass");
	}
	
	
	private static List<Link> getCorLinks(Map<Integer, List<DPoint>> colMap){
		List<Link> corLinks = new ArrayList<Link>();
		SimpleUnion u;
		
		int colNum;
		List<DPoint> colPoints;
		int i;
		int j;
		double rightBoundary;
		
		int edges;
		
		for(Entry<Integer, List<DPoint>> col : colMap.entrySet()){
			colNum = col.getKey().intValue();
			edges = 0;
			
			rightBoundary = ((double)colNum) * G + Constants.BORDERLESS_COL_WIDTH;
			
			colPoints = col.getValue();
			u = new SimpleUnion();
			
			for(DPoint f : colPoints){
				u.add(f);
			}
			
			
			for(i = 1; i < colPoints.size(); i++){
				for(j = 0; j < i; j++){
					if(colPoints.get(i).withinRange(colPoints.get(j), Lmax_SQUARED)){
						u.union(colPoints.get(i), colPoints.get(j));
					}
				}
			}
			
			for(Entry<DPoint, DPoint> e : u.parentMap.entrySet()){
				if(e.getKey().x >= rightBoundary){
					corLinks.add(new Link(e.getKey(), e.getValue()));
				}
			}
			
			
		}
		
		Collections.sort(corLinks);
		return corLinks;
	}
	
	private static Map<Integer, List<DPoint>> getColMap(List<DPoint> allPoints){
		Map<Integer, List<DPoint>> colMap = new HashMap<Integer, List<DPoint>>();
		int colNum;
		double offset;
		
		for(DPoint f : allPoints){
			colNum = (int)(f.x / G);
			offset = f.x - ((double)colNum) * G;
			addToColumn(colMap, colNum, f);
			if(offset > G - Constants.Lmax){
				addToColumn(colMap, colNum + 1, f);
			}
			
		}
		
		return colMap;
	}
	
	private static int getCorEdgeCount(List<DPoint> allPoints){
		int edgeCount = 0;
		int i;
		int j;
		
		for(i = 1; i < allPoints.size(); i++){
			for(j = 0; j < i; j++){
				if(allPoints.get(i).withinRange(allPoints.get(j), Lmax_SQUARED)){
					edgeCount++;
				}
			}
		}
		return edgeCount;
	}
	
	private static void addToColumn(Map<Integer, List<DPoint>> colMap, int colNum, DPoint f){
		if(colMap.containsKey(colNum)){
			colMap.get(colNum).add(f);
		}else{
			List<DPoint> list = new ArrayList<DPoint>();
			list.add(f);
			colMap.put(colNum, list);
		}
		
	}

	private static List<Link> getCalcLinks(String path) throws Exception{
		List<Link> calcLinks = new ArrayList<Link>();
		
		BufferedReader dataReader = new BufferedReader(new FileReader(path));

		
		String line = dataReader.readLine();
		String[] majorSplit;
		String[] minorSplit;
		double x;
		double y;
		
		Link l;
		
		while(line != null){
			majorSplit = line.split("\t");
			minorSplit = majorSplit[0].split(" ");

			l = new Link();
			
			x = Double.parseDouble(minorSplit[0]);
			y = Double.parseDouble(minorSplit[1]);

			l.child = new DPoint(x,y);

			minorSplit = majorSplit[1].split(" ");
			
			x = Double.parseDouble(minorSplit[0]);
			y = Double.parseDouble(minorSplit[1]);

			l.parent = new DPoint(x,y);
			
			calcLinks.add(l);
			line = dataReader.readLine();
		}
		
		Collections.sort(calcLinks);
		
		return calcLinks;
	}
	
	private static List<DPoint> getPointsFromDataFiles(String path) throws Exception{
		List<DPoint> allPoints = new ArrayList<DPoint>();
		
		BufferedReader dataReader = new BufferedReader(new FileReader(path));

		
		String line = dataReader.readLine();
		String[] split;
		double x;
		double y;
		double weight;
		
		while(line != null){
			split = line.split(" ");
			x = Double.parseDouble(split[0]);
			y = Double.parseDouble(split[1]);
			weight = Double.parseDouble(split[2]);
			
			if(Constants.MIN_WEIGHT <= weight && Constants.MAX_WEIGHT > weight){
				allPoints.add(new DPoint(x, y));
			}
			
			line = dataReader.readLine();
		}
		
		return allPoints;
	}
	
}

class DPoint{
	double x;
	double y;
	
	DPoint(double x, double y){
		this.x = x;
		this.y = y;
	}

	public boolean equals(Object otherOb){
		if(otherOb == null || (! (otherOb instanceof DPoint))) return false;
		DPoint other = (DPoint)otherOb;
		
		return x == other.x && y == other.y;
	}
	
	@Override
	public int hashCode(){
		return new Double(x).hashCode() ^ new Double(y).hashCode();
	}
	
	public boolean withinRange(DPoint other, double rangeSquared){
		return (x - other.x) * (x - other.x) + (y - other.y) * (y - other.y) <= rangeSquared;
	}
	
	public boolean isLessThanX(DPoint other){
		if(x < other.x) return true;
		if(x > other.x) return false;
		if(y < other.y) return true;
		if(y > other.y) return false;
		return false;
	}
	
}


class Link implements Comparable<Link>{
	DPoint child;
	DPoint parent;
	int edgeCount;
	
	Link(){
		edgeCount = -1;
	}
	
	Link(DPoint child, DPoint parent){
		this.child = child;
		this.parent = parent;
		edgeCount = -1;
	}
	
	Link(int edgeCount){
		this.edgeCount = edgeCount;
	}
	
	public boolean equals(Object otherOb){
		if(otherOb == null || (! (otherOb instanceof Link))) return false;
		Link other = (Link)otherOb;
		
		return child.equals(other.child) && parent.equals(other.parent) && edgeCount == other.edgeCount;
		
	}
	
	public String toString(){
		if(edgeCount == -1)	return child.x + " " + child.y + "\t" + parent.x + " " + parent.y;
		else if(edgeCount == 1) return "1 edge";
		else return edgeCount + " edges";
	}

	@Override
	public int compareTo(Link arg0) {
		
		if(edgeCount > -1){
			if(arg0.edgeCount > -1)	return arg0.edgeCount - edgeCount;
			return -1;
		}else if(arg0.edgeCount > -1) return 1;
		
		
		if(child.isLessThanX(arg0.child)) return -1;
		if(arg0.child.isLessThanX(child)) return 1;
		if(parent.isLessThanX(arg0.child)) return -1;
		if(arg0.child.isLessThanX(parent)) return 1;
		return 0;
	}
}

class SimpleUnion{
	
	
	Map<DPoint, DPoint> parentMap;
	Map<DPoint, List<DPoint>> childMap; 
	
	SimpleUnion(){
		parentMap = new HashMap<DPoint, DPoint>();
		childMap = new HashMap<DPoint, List<DPoint>>();
	}
	
	void add(DPoint f){
		ArrayList<DPoint> list = new ArrayList<DPoint>();
		list.add(f);
		parentMap.put(f, f);
		childMap.put(f, list);
	}
	
	void union(DPoint f1, DPoint f2){
		DPoint p1 = parentMap.get(f1);
		DPoint p2 = parentMap.get(f2);
		
		if(p1 == p2) return;
		
		DPoint newParent;
		DPoint oldParent;
		
		List<DPoint> biggerList;
		List<DPoint> deadList;
		
		if(p1.isLessThanX(p2)){
			newParent = p1;
			oldParent = p2;
		}else{
			newParent = p2;
			oldParent = p1;
		}
		
		deadList = childMap.get(oldParent);
		biggerList = childMap.get(newParent);
		
		for(DPoint f : deadList){
			parentMap.put(f, newParent);
			biggerList.add(f);
		}
		
		childMap.remove(oldParent);
		
	}
	
	DPoint find(DPoint f){
		return parentMap.get(f);
	}
	
	
	void printState(){
		System.out.println(parentMap.size() + " points");
		System.out.println(childMap.size() + " unions\n");
		
		StringBuilder sb;
		for(Entry<DPoint, List<DPoint>> e : childMap.entrySet()){
			sb = new StringBuilder(String.valueOf(e.getKey().x)).append(": ");
			for(DPoint f : e.getValue()){
				sb.append(f.x);
				sb.append(", ");
			}
			System.out.println(sb.toString());
		}
		
		System.out.println("\n");
		
		for(Entry<DPoint, DPoint> e : parentMap.entrySet()){
			sb = new StringBuilder(String.valueOf(e.getKey().x));
			sb.append(" -> ");
			sb.append(e.getValue().x);
			System.out.println(sb.toString());
		}
		
		
	}
	
}